package com.it.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;

/**
 * 本文hbase为2.x版本，教程是1.x版本，故api不相同
 */
public class HBaseImportTest /*extends Thread*/ {
    private Connection connection;
    private Table table;

    public HBaseImportTest() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "cdh1,cdh2,cdh3");
        try {
            connection = ConnectionFactory.createConnection(conf);
            createTable("default", "user_action","f1");
            table = connection.getTable(TableName.valueOf("default", "user_action"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建NameSpace
     */
    public void createNameSpace(String nameSpace) throws IOException {
        // 基本的判空操作
        if (nameSpace == null || nameSpace.equals("")) {
            System.err.println("nameSpace名字不能为空");
            return;
        }
        // 获取Admin对象
        Admin admin = connection.getAdmin();
        NamespaceDescriptor.Builder builder = NamespaceDescriptor.create(nameSpace);
        NamespaceDescriptor namespaceDescriptor = builder.build();
        try {
            // 调用方法
            admin.createNamespace(namespaceDescriptor);
            System.out.println(nameSpace + " 创建成功");
        } catch (NamespaceExistException e) {
            System.err.println(nameSpace + " 已经存在");
        } finally {
            admin.close();
        }
    }

    public static void main(String[] args) throws Exception {
       /* if (args.length == 0) {       //第一个参数是该jar所使用的类，第二个参数是数据集所存放的路径
            throw new Exception("You must set input path!");
        }

        String fileName = args[args.length - 1];  //输入的文件路径是最后一个参数*/
        String fileName = "E:\\code\\java\\ImportHBase\\data\\user_action.output";  //本地调试的文件路径

        HBaseImportTest test = new HBaseImportTest();
        test.importLocalFileToHBase(fileName);
    }

    public void importLocalFileToHBase(String fileName) {
        long st = System.currentTimeMillis();
        BufferedReader br = null;
        try {
            br = new BufferedReader(new InputStreamReader(new FileInputStream(
                    fileName)));
            String line = null;
            int count = 0;
            while ((line = br.readLine()) != null) {
                count++;
                put(line);// 写数据
                if (count % 10000 == 0)
                    System.out.println(count);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {

            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            try {
                table.close(); // must close the client
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
        long en2 = System.currentTimeMillis();
        System.out.println("Total Time: " + (en2 - st) + " ms");
    }

    // 新增一条
    @SuppressWarnings("deprecation")
    public void put(String line) throws IOException {
        String[] arr = line.split("\t", -1);
        String[] column = {"id", "uid", "item_id", "behavior_type", "item_category", "date", "province"};

        if (arr.length == 7) {
            Put put = new Put(Bytes.toBytes(arr[0]));// rowkey
            for (int i = 1; i < arr.length; i++) {
                put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes(column[i]), Bytes.toBytes(arr[i]));
            }
            table.put(put); // put to server
        }
    }

    // 查询一条
    public void get(String rowkey, String columnFamily, String column,
                    int versions) throws IOException {
        long st = System.currentTimeMillis();

        Get get = new Get(Bytes.toBytes(rowkey));
        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));

        Scan scanner = new Scan(get);
        scanner.setMaxVersions(versions);

        ResultScanner rsScanner = table.getScanner(scanner);

        for (Result result : rsScanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                System.out.println(Bytes.toStringBinary(cell.getValueArray()) + "\t"
                        + cell.getTimestamp()); // mid + time
            }
        }
        rsScanner.close();

        long en2 = System.currentTimeMillis();
        System.out.println("Total Time: " + (en2 - st) + " ms");
    }

    /**
     * 判断表是否存在
     */
    public boolean existsTable(String nameSpaceName, String tableName) throws IOException {
        Admin admin = connection.getAdmin();
        return admin.tableExists(TableName.valueOf(nameSpaceName, tableName));
    }


    /**
     * 创建table
     */
    public void createTable(String nameSpaceName, String tableName, String... cfs) throws IOException {

        if (existsTable(nameSpaceName, tableName)) {
            System.err.println((nameSpaceName == null || nameSpaceName.equals("") ? "default" : nameSpaceName) + ":" + tableName + "表已经存在");
            return;
        }

        Admin admin = connection.getAdmin();

        TableDescriptorBuilder tableDescriptorBuilder =
                TableDescriptorBuilder.newBuilder(TableName.valueOf(nameSpaceName, tableName));

        if (cfs == null || cfs.length < 1) {
            System.err.println("至少指定一个列族");
            return;
        }

        for (String cf : cfs) {
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
            columnFamilyDescriptorBuilder.setMaxVersions(5);
            ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();

            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
        }


        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();


        admin.createTable(tableDescriptor);

        admin.close();
    }

}
